package raft

//
// this is an outline of the API that raft must expose to
// the service (or tester). see comments below for
// each of these functions for more details.
//
// rf = Make(...)
//   create a new Raft server.
// rf.Start(command interface{}) (index, term, isleader)
//   start agreement on a new log entry
// rf.GetState() (term, isLeader)
//   ask a Raft for its current term, and whether it thinks it is leader
// ApplyMsg
//   each time a new entry is committed to the log, each Raft peer
//   should send an ApplyMsg to the service (or tester)
//   in the same server.
//

import (
	"bytes"
	"context"
	"fmt"
	"log"
	"math/rand"
	"sort"
	"time"

	"mit-6.824/src/labgob"
	"mit-6.824/src/labrpc"
)

// import "bytes"
// import "../labgob"

func ack(done chan<- bool) {
	done <- true
}

//
// as each Raft peer becomes aware that successive log entries are
// committed, the peer should send an ApplyMsg to the service (or
// tester) on the same server, via the applyCh passed to Make(). set
// CommandValid to true to indicate that the ApplyMsg contains a newly
// committed log entry.
//
// in Lab 3 you'll want to send other kinds of messages (e.g.,
// snapshots) on the applyCh; at that point you can add fields to
// ApplyMsg, but set CommandValid to false for these other uses.
//
type ApplyMsg struct {
	CommandValid bool
	Command      interface{}
	CommandIndex int
}

// Raft configurations

const (
	HEARTBEAT_SPAN        = 100
	ELECTION_TIMEOUT_MIN  = 5 * HEARTBEAT_SPAN / 2
	ELECTION_TIMEOUT_SPAN = ELECTION_TIMEOUT_MIN
	RPC_TIMEOUT           = 10000
	FETCH_TIMEOUT         = 5
)

// Raft peer state
type RaftState int

const (
	FOLLOWER RaftState = iota
	CANDIDATE
	LEADER
)

func (s RaftState) String() string {
	switch s {
	case FOLLOWER:
		return "f"
	case CANDIDATE:
		return "C"
	case LEADER:
		return "L"
	default:
		return "-"
	}
}

//
// A Go object implementing a single Raft peer.
//
type Raft struct {
	ctx    context.Context
	cancel context.CancelFunc

	// mu        sync.Mutex          // Lock to protect shared access to this peer's state
	// dead      int32               // set by Kill()

	peers     []*labrpc.ClientEnd // RPC end points of all peers
	persister *Persister          // Object to hold this peer's persisted state
	me        int                 // this peer's index into peers[]
	applyCh   chan ApplyMsg

	majority int

	// Your data here (2A, 2B, 2C).
	// Look at the paper's Figure 2 for a description of what
	// state a Raft server must maintain.

	// persistent states
	currentTerm int
	votedFor    int // -1 for nil

	// volatile states
	state      RaftState
	matchIndex []int

	// the log
	log *LogKeeper

	// message passing
	clientGetStateCh chan ClientGetStateEvent
	clientStartCh    chan ClientStartEvent
	requestVoteCh    chan RequestVoteEvent
	appendEntriesCh  chan AppendEntriesEvent
}

// return currentTerm and whether this server
// believes it is the leader.

type ClientGetStateReply struct {
	term     int
	isLeader bool
}

type ClientGetStateEvent struct {
	reply *ClientGetStateReply
	done  chan bool
}

func (rf *Raft) GetState() (int, bool) {
	// Your code here (2A).
	var result ClientGetStateReply

	rf.debug("GetState")
	done := make(chan bool)

	select {
	case rf.clientGetStateCh <- ClientGetStateEvent{reply: &result, done: done}:
	case <-rf.ctx.Done():
	}

	select {
	case <-done:
	case <-rf.ctx.Done():
	}

	return result.term, result.isLeader
}

func (rf *Raft) replyGetState(e ClientGetStateEvent) {
	rf.debug("replyGetState")
	e.reply.term = rf.currentTerm
	e.reply.isLeader = rf.state == LEADER
	ack(e.done)
}

//
// save Raft's persistent state to stable storage,
// where it can later be retrieved after a crash and restart.
// see paper's Figure 2 for a description of what should be persistent.
//
func (rf *Raft) persist() {
	// Your code here (2C).
	// Example:
	// w := new(bytes.Buffer)
	// e := labgob.NewEncoder(w)
	// e.Encode(rf.xxx)
	// e.Encode(rf.yyy)
	// data := w.Bytes()
	// rf.persister.SaveRaftState(data)

	rf.debug("rf.persist")

	w := &bytes.Buffer{}
	e := labgob.NewEncoder(w)

	if err := e.Encode(rf.currentTerm); err != nil {
		panic(err)
	}

	if err := e.Encode(rf.votedFor); err != nil {
		panic(err)
	}

	if err := rf.log.Save(e); err != nil {
		panic(err)
	}

	rf.persister.SaveRaftState(w.Bytes())
}

//
// restore previously persisted state.
//
func (rf *Raft) readPersist(data []byte) {
	if data == nil || len(data) < 1 { // bootstrap without any state?
		return
	}
	// Your code here (2C).
	// Example:
	// r := bytes.NewBuffer(data)
	// d := labgob.NewDecoder(r)
	// var xxx
	// var yyy
	// if d.Decode(&xxx) != nil ||
	//    d.Decode(&yyy) != nil {
	//   error...
	// } else {
	//   rf.xxx = xxx
	//   rf.yyy = yyy
	// }

	rf.debug("rf.readPersist")

	r := bytes.NewBuffer(data)
	d := labgob.NewDecoder(r)

	var newCurrentTerm int
	if err := d.Decode(&newCurrentTerm); err != nil {
		panic(err)
	} else {
		rf.currentTerm = newCurrentTerm
	}

	var newVotedFor int
	if err := d.Decode(&newVotedFor); err != nil {
		panic(err)
	} else {
		rf.votedFor = newVotedFor
	}

	if err := rf.log.Load(d); err != nil {
		panic(err)
	}
}

//
// example RequestVote RPC arguments structure.
// field names must start with capital letters!
//
type RequestVoteArgs struct {
	// Your data here (2A, 2B).
	Term         int
	CandidateID  int
	LastLogTerm  int
	LastLogIndex int
}

//
// example RequestVote RPC reply structure.
// field names must start with capital letters!
//
type RequestVoteReply struct {
	// Your data here (2A).
	Term        int
	VoteGranted bool
}

type RequestVoteEvent struct {
	args  *RequestVoteArgs
	reply *RequestVoteReply
	done  chan bool
}

func (u *RequestVoteArgs) String() string {
	return fmt.Sprintf("c=[%d/%d],l=(%d/%d)",
		u.CandidateID, u.Term, u.LastLogIndex, u.LastLogTerm)
}

//
// example RequestVote RPC handler.
//
func (rf *Raft) RequestVote(args *RequestVoteArgs, reply *RequestVoteReply) {
	// Your code here (2A, 2B).

	// how can we get nil args...
	if args == nil || reply == nil {
		return
	}

	rf.debug("received RequestVote from [%d/%d]: %s", args.CandidateID, args.Term, args)
	done := make(chan bool)
	rf.requestVoteCh <- RequestVoteEvent{args: args, reply: reply, done: done}
	<-done
}

func (rf *Raft) denyRequestVote(e RequestVoteEvent) {
	rf.debug("denyRequestVote [%d/%d]: %s", e.args.CandidateID, e.args.Term, e.args)
	e.reply.Term = rf.currentTerm
	e.reply.VoteGranted = false
	ack(e.done)
}

// replayXXX puts e back to channel and never blocks the caller.
func (rf *Raft) replayRequestVote(e RequestVoteEvent) {
	go func() {
		rf.requestVoteCh <- e
	}()
}

//
// example code to send a RequestVote RPC to a server.
// server is the index of the target server in rf.peers[].
// expects RPC arguments in args.
// fills in *reply with RPC reply, so caller should
// pass &reply.
// the types of the args and reply passed to Call() must be
// the same as the types of the arguments declared in the
// handler function (including whether they are pointers).
//
// The labrpc package simulates a lossy network, in which servers
// may be unreachable, and in which requests and replies may be lost.
// Call() sends a request and waits for a reply. If a reply arrives
// within a timeout interval, Call() returns true; otherwise
// Call() returns false. Thus Call() may not return for a while.
// A false return can be caused by a dead server, a live server that
// can't be reached, a lost request, or a lost reply.
//
// Call() is guaranteed to return (perhaps after a delay) *except* if the
// handler function on the server side does not return.  Thus there
// is no need to implement your own timeouts around Call().
//
// look at the comments in ../labrpc/labrpc.go for more details.
//
// if you're having trouble getting RPC to work, check that you've
// capitalized all field names in structs passed over RPC, and
// that the caller passes the address of the reply struct with &, not
// the struct itself.
//
func (rf *Raft) sendRequestVote(server int, args *RequestVoteArgs, reply *RequestVoteReply) bool {
	rf.debug("sendRequestVote to [%d/*]: %s", server, args)
	ok := rf.peers[server].Call("Raft.RequestVote", args, reply)

	if ok {
		rf.debug("sendRequestVote to [%d/%d] done: %s", server, reply.Term, args)
	} else {
		rf.debug("sendRequestVote to [%d/*] failed: %s", server, args)
	}

	return ok
}

// AppendEntries RPC

type AppendEntriesArgs struct {
	Term         int
	LeaderID     int
	LeaderCommit int
	PrevLogTerm  int
	PrevLogIndex int
	Entries      LogSlice
}

func (u *AppendEntriesArgs) String() string {
	prefix := ""
	entries := u.Entries
	if len(entries) >= LOG_DUMP_LIMIT {
		prefix = "..."
		entries = entries[len(entries)-LOG_DUMP_LIMIT:]
	}

	return fmt.Sprintf("l=[%d/%d],p=(%d/%d),c=%d,e=%s%s",
		u.LeaderID, u.Term,
		u.PrevLogIndex, u.PrevLogTerm,
		u.LeaderCommit,
		prefix, entries,
	)
}

type AppendEntriesReply struct {
	Term           int
	ConflictTerm   int
	BacktrackIndex int
	Success        bool
}

type AppendEntriesEvent struct {
	args  *AppendEntriesArgs
	reply *AppendEntriesReply
	done  chan bool
}

func (rf *Raft) AppendEntries(args *AppendEntriesArgs, reply *AppendEntriesReply) {
	if args == nil || reply == nil {
		return
	}

	rf.debug("received AppendEntries %s", args)
	done := make(chan bool)
	rf.appendEntriesCh <- AppendEntriesEvent{args: args, reply: reply, done: done}
	<-done
}

func (rf *Raft) denyAppendEntries(e AppendEntriesEvent) {
	rf.debug("denyAppendEntries %s", e.args)
	e.reply.Term = rf.currentTerm
	e.reply.ConflictTerm = -1
	e.reply.BacktrackIndex = rf.getLastIndex()
	e.reply.Success = false
	ack(e.done)
}

func (rf *Raft) replayAppendEntries(e AppendEntriesEvent) {
	go func() {
		rf.appendEntriesCh <- e
	}()
}

func (rf *Raft) sendAppendEntries(server int, args *AppendEntriesArgs, reply *AppendEntriesReply) bool {
	rf.debug("sendAppendEntries to [%d/*]: %s", server, args)
	ok := rf.peers[server].Call("Raft.AppendEntries", args, reply)

	if ok {
		rf.debug("sendAppendEntries to [%d/%d] done: %s", server, reply.Term, args)
	} else {
		rf.debug("sendAppendEntries to [%d/*] failed: %s", server, args)
	}

	return ok
}

//
// the service using Raft (e.g. a k/v server) wants to start
// agreement on the next command to be appended to Raft's log. if this
// server isn't the leader, returns false. otherwise start the
// agreement and return immediately. there is no guarantee that this
// command will ever be committed to the Raft log, since the leader
// may fail or lose an election. even if the Raft instance has been killed,
// this function should return gracefully.
//
// the first return value is the index that the command will appear at
// if it's ever committed. the second return value is the current
// term. the third return value is true if this server believes it is
// the leader.
//
func (rf *Raft) Start(command interface{}) (int, int, bool) {
	// Your code here (2B).

	done := make(chan ClientStartReply)

	select {
	case rf.clientStartCh <- ClientStartEvent{command, done}:
		select {
		case reply := <-done:
			rf.debug("Start %#v @index=%d, term=%d, ok=%v",
				command, reply.index, reply.term, reply.isLeader)
			return reply.index, reply.term, reply.isLeader
		case <-rf.ctx.Done():
			return -1, -1, false
		}
	case <-rf.ctx.Done():
		return -1, -1, false
	}
}

type ClientStartReply struct {
	index    int
	term     int
	isLeader bool
}

type ClientStartEvent struct {
	command interface{}
	done    chan ClientStartReply
}

func (rf *Raft) replyStart(e ClientStartEvent) {
	if rf.state == LEADER {
		index := rf.log.AppendNew(e.command)
		rf.matchIndex[rf.me] = index
		rf.persist()

		// seems that MIT's index is started at 1.
		// make them happy.
		e.done <- ClientStartReply{index + 1, rf.currentTerm, true}
	} else {
		e.done <- ClientStartReply{-1, -1, false}
	}
}

//
// the tester doesn't halt goroutines created by Raft after each test,
// but it does call the Kill() method. your code can use killed() to
// check whether Kill() has been called. the use of atomic avoids the
// need for a lock.
//
// the issue is that long-running goroutines use memory and may chew
// up CPU time, perhaps causing later tests to fail and generating
// confusing debug output. any goroutine with a long-running loop
// should call killed() to check whether it should stop.
//
func (rf *Raft) Kill() {
	// atomic.StoreInt32(&rf.dead, 1)
	// Your code here, if desired.
	rf.cancel()
}

func (rf *Raft) killed() bool {
	// z := atomic.LoadInt32(&rf.dead)
	// return z == 1
	return rf.ctx.Err() != nil
}

//
// the service or tester wants to create a Raft server. the ports
// of all the Raft servers (including this one) are in peers[]. this
// server's port is peers[me]. all the servers' peers[] arrays
// have the same order. persister is a place for this server to
// save its persistent state, and also initially holds the most
// recent saved state, if any. applyCh is a channel on which the
// tester or service expects Raft to send ApplyMsg messages.
// Make() must return quickly, so it should start goroutines
// for any long-running work.
//
func Make(peers []*labrpc.ClientEnd, me int,
	persister *Persister, applyCh chan ApplyMsg) *Raft {
	log.SetFlags(log.Lmicroseconds)

	rf := &Raft{}
	rf.ctx, rf.cancel = context.WithCancel(context.Background())

	rf.peers = peers
	rf.persister = persister
	rf.me = me
	rf.applyCh = applyCh

	rf.majority = len(peers)/2 + 1

	// Your initialization code here (2A, 2B, 2C).

	// default values for states when bootstrap.
	rf.currentTerm = 0
	rf.votedFor = -1
	rf.log = NewLogger(rf.ctx, rf.majority, rf.applyCh, rf.debug)

	// initialize from state persisted before a crash
	rf.readPersist(persister.ReadRaftState())

	// start working goroutine.
	rf.state = FOLLOWER
	rf.matchIndex = make([]int, len(peers))

	// peer RPC channels
	rf.requestVoteCh = make(chan RequestVoteEvent)
	rf.appendEntriesCh = make(chan AppendEntriesEvent)

	// client request channels
	rf.clientGetStateCh = make(chan ClientGetStateEvent)
	rf.clientStartCh = make(chan ClientStartEvent)

	go rf.main(rf.ctx)

	return rf
}

// get a election timer
func newElectionTimer() (timer *time.Timer, timeout int) {
	timeout = rand.Intn(ELECTION_TIMEOUT_SPAN) + ELECTION_TIMEOUT_MIN
	timer = time.NewTimer(time.Duration(timeout) * time.Millisecond)
	return
}

// reset states to follower
func (rf *Raft) resetToFollower(term int) {
	if rf.state != FOLLOWER {
		rf.debug("downgrade to a follower")
	}

	rf.currentTerm = term
	rf.votedFor = -1
	rf.state = FOLLOWER
	rf.persist()

	rf.log.UpdateTerm(term)
}

// set state to candidate and start a new term
func (rf *Raft) startElection() {
	if rf.state == CANDIDATE {
		rf.debug("restart election")
	} else {
		rf.debug("start election")
	}

	rf.currentTerm += 1
	rf.state = CANDIDATE
	rf.persist()
}

// return true if currentTerm updates.
func (rf *Raft) checkTerm(term int) bool {
	if term > rf.currentTerm {
		rf.debug("term %d → %d", rf.currentTerm, term)
		rf.resetToFollower(term)
		return true
	} else {
		return false
	}
}

// return true if a valid new leader was elected.
func (rf *Raft) checkTermWithEq(term int) bool {
	if term >= rf.currentTerm {
		rf.debug("lost election: term %d → %d", rf.currentTerm, term)
		rf.resetToFollower(term)
		return true
	} else {
		return false
	}
}

// try to grant vote to e.
func (rf *Raft) tryGrantVote(e RequestVoteEvent) (granted, first bool) {
	granted = rf.checkLogContained(e.args.LastLogTerm, e.args.LastLogIndex) &&
		(rf.votedFor == -1 || rf.votedFor == e.args.CandidateID)

	e.reply.Term = rf.currentTerm
	e.reply.VoteGranted = granted

	if granted {
		rf.debug("grant vote to [%d/%d]", e.args.CandidateID, e.args.Term)

		if rf.votedFor == -1 {
			first = true
		}

		rf.votedFor = e.args.CandidateID
		rf.persist()
	} else {
		rf.debug("refuse to vote [%d/%d]", e.args.CandidateID, e.args.Term)
	}

	ack(e.done)
	return
}

// check whether the term of log at index matches argument term.
func (rf *Raft) checkLogMatched(term, index int) bool {
	return rf.getLogTerm(index) == term
}

// check whether a log with term and index contains last log.
func (rf *Raft) checkLogContained(term, index int) bool {
	lastLog := rf.log.GetLastLog()
	return lastLog == nil || term > lastLog.Term ||
		(term == lastLog.Term && index >= lastLog.Index)
}

// initialize leader-related states
func (rf *Raft) initLeader() {
	rf.debug("initLeader")
	rf.state = LEADER
	for i := range rf.matchIndex {
		rf.matchIndex[i] = -1
	}
}

// Raft peer's working goroutine.
func (rf *Raft) main(ctx context.Context) {
	rf.debug("hello, world!")

	for !rf.killed() {
		rf.log.UpdateTerm(rf.currentTerm)

		switch rf.state {
		case FOLLOWER:
			rf.runFollower(ctx)
		case CANDIDATE:
			rf.runCandidate(ctx)
		case LEADER:
			rf.runLeader(ctx)
		}
	}
}

// get the last index of log.
// return -1 if there's no log.
func (rf *Raft) getLastIndex() int {
	lastLog := rf.log.GetLastLog()
	if lastLog == nil {
		return -1
	} else {
		return lastLog.Index
	}
}

// return the term of log at index.
// if there's no log at index, return -1
func (rf *Raft) getLogTerm(index int) int {
	log := rf.log.GetLog(index)
	if log == nil {
		return -1
	} else {
		return log.Term
	}
}

func (rf *Raft) search(pred func(int) bool) int {
	return rf.log.Search(pred)
}

func (rf *Raft) getSortedMatchIndex() (minIndex int, majorIndex int, maxIndex int) {
	sorted := make([]int, len(rf.peers))
	copy(sorted, rf.matchIndex)
	sort.Ints(sorted)
	rf.debug("sorted: %v", sorted)

	minIndex = sorted[0]
	majorIndex = sorted[rf.majority-1]
	maxIndex = sorted[len(rf.peers)-1]
	return
}
